チャットUIは現代の職場を席巻しています。制作チームはスレッドで連携し、承認はメールで行われ、LLM搭載のアシスタントが日々のオペレーションの一部になりつつあります。
本当の問題は、適切な統合です。たとえば「レビュー準備完了」のショットを知らせるメッセージは、理想的には、スーパーバイザーがそのショットを承認し、正しいユーザーのもとでKitsuのステータスを更新できるはずです。しかしそのためには、小さなバックエンドサービス、Kitsuへの安全なAPI接続、そしてチャットユーザーとKitsuユーザーを確実に対応付ける仕組みが必要になります。朗報として、あなたはすでにKitsuでそれを実現できます！
シンプルな出発点は、/hello のような1つのコマンドを持つTelegramボットです。ボットは最初にチャットユーザーをKitsuアカウントへ一度だけリンクし、その後はAPIを通じて返信し、チャット上に表示します。Kitsuで何かイベントが起きるたびに、ボットがあなたに通知します。この小さな統合がコンセプトを証明しており、この記事ではまさにそれを作っていきます。
カスタム・メッセージング統合が必要な理由
カスタム・メッセージング統合は、単一の真実の情報源（single source of truth）を中心にコミュニケーションを集約します。スーパーバイザーがタスクのステータス変更に関するメールを転送する代わりに、更新は自動的に該当するチームのチャネルへプッシュできます。たとえば、照明タスクがKitsuで「撮り直し（retake）」に切り替わると、照明のTelegramグループはショット名、担当者、締切を含む構造化されたメッセージを即座に受け取ります。制作トラッカーは先回りして動くようになります。
生のデータベースイベントを読みやすい要約に再構成すると、ユーザー体験は向上します。アーティストが、何が変わったのかを理解するためにアクティビティログを掘り下げる必要はありません。Telegramのチャンネルに送られる日次ダイジェストには、承認、新しい割り当て、今後の締切が平易な言葉でまとめられます。このダイジェストはKitsu APIから直接生成でき、毎晩自動で配信することで、制作データを“人が実際に読む”ものへ変えられます。
ただし、このアプローチの本当の効果が発揮されるのは、自動化の場面です。メッセージングプラットフォームは軽量なコマンドインターフェースとして機能します。たとえば、Telegramで「/late_shots」と入力するコーディネーターは、Kitsuへクエリを投げて、期限超過のタスク一覧を即座に受け取れます。「/assign SH010 alice」とタイプするリードは、バックエンド呼び出しをトリガーしてKitsuの割り当てを更新できます。チャットは、制作データベースのための運用上の“操作面”になります。
とはいえ、先ほど言ったとおり、まずはKitsuと連携するTelegramボットから始めましょう。
1. 新しいTelegramボットを作成する
まず、Telegram上で専用のボットを作成します。ボットを分離しておくことで、資格情報（クレデンシャル）をきれいに保て、制作環境へ引き渡す際に将来のセキュリティ上の面倒が増えるのを避けられます。
Telegramを開き、他のボットを管理するための公式ボットであるBotFatherを検索します。
チャットを開始し、/newbot を送信します。手順はシンプルです。「Kitsu Notifications」のような人が読める名前を入力し、その後に kitsu_pipeline_bot のような固有のユーザー名を設定します。ユーザー名は「bot」で終わる必要があり、かつグローバルに一意であるため、スタジオ環境ではいくつか試すことになるでしょう。
BotFatherはAPIトークンを返します。このトークンをSlackに貼り付けたり、Gitへコミットするための“便利な文字列”として扱わないでください。環境設定システムに保存します。このトークンが漏れると、誰でも制作ボットとしてメッセージを送れます。制作担当者がスパムを受け取り始めると、あっという間に“面白い”状態から“致命的”な状態へ変わります。
これをKitsuのイベントシステムに接続する前に、トークンを手動で検証しましょう。
Telegram内で、作成したばかりのボットをユーザー名で検索し、そのボットとの会話を開始します。Telegramがチャットを登録できるように、シンプルに「/start」を送信してください。
クライアント（チャット）IDを取得するには、トークンを使ってcurlで getUpdates エンドポイントを呼び出します。例えば：
curl https://api.telegram.org/bot<TOKEN>/getUpdates
レスポンスには、chatオブジェクトと idフィールドを含むJSONペイロードが入ります。この数値IDが、統合がターゲットにするものです。実際のパイプラインのシナリオでは、個人ユーザーというより、スーパーバイザーのグループのチャットIDであることもあります。
次に、アウトバウンドのメッセージ送信を直接テストします。curlで自分宛てにメッセージを送ってください：
curl -X POST https://api.telegram.org/bot<TOKEN>/sendMessage -d chat_id=<CHAT_ID> -d text="Kitsu integration test"
メッセージがTelegramに表示されれば、トークンとチャットIDは有効です。この手動検証のステップは、同じ呼び出しをKitsuのイベントフックに組み込んだときに、何かが静かに失敗するようなケースで、後からデバッグに数時間を費やすのを防いでくれます。
ボットが検証できたら、次はKitsuのイベントシステムに接続します。たとえば、新しいアセットが作成されたら、スーパーバイザーのTelegramグループにメッセージが自動でプッシュされるようにします。
curlでテストしたのとまったく同じ sendMessage エンドポイントが、Kitsuによってトリガーされる小さなサービス、またはサーバーレス関数の一部になります。
2. Kitsuのイベントリスナーを設定する
次に、Kitsuからのリアルタイムイベントを購読する必要があります。目的はシンプルです。制作データが変わる瞬間に反応します。
Kitsuの zou Python SDKを使ってウェブソケット接続を開き、タスク更新イベントをリッスンできます。
たとえば、Kitsuのイベントストリームに接続し、アセット作成イベントをフィルタリングします：
import gazugazu.set_host("http://localhost:80/api") gazu.set_event_host("http://localhost:80/api") gazu.log_in("admin@example.com", "mysecretpassword")
def my_callback(data): print("Asset created %s" % data"asset_id")
event_client = gazu.events.init() gazu.events.add_listener(event_client, "asset:new", my_callback) gazu.events.run_client(event_client)
gazu ライブラリを使って、http://localhost:80/api のローカルホスト上にあるKitsu APIサーバーへ接続し、指定された管理者認証情報で認証したうえで、リアルタイムイベントをリッスンします。
この抜粋では、イベントがトリガーされるたびに新しく作成されたアセットのIDを出力するコールバック関数 my_callback を定義しています。
gazu.events.init() でイベントクライアントを初期化した後、スクリプトは "asset:new" イベントをリッスンするためにコールバックを登録します（このイベントは、システム内で新しいアセットが作成されるたびに発火します）。
gazu.events.run_client(event_client) はイベントループを開始し、Kitsuに新しいアセットが追加されるたびにコールバックが実行され、その asset_id が出力される状態を維持します。
3. メッセージを送るためにTelegram APIを使う
イベントが流れてくる状態になったら、先ほどのテストと同様に、Telegramの sendMessage エンドポイントでメッセージを外へ押し出します。APIは、HTTP POSTで、ボットトークン、チャットID、テキストのペイロードを含めるだけです。
それを小さなユーティリティ関数にまとめます：
import requests import osTELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN') TELEGRAM_CHAT_ID = os.getenv('TELEGRAM_CHAT_ID')
def send_telegram_message(text): url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" payload = { "chat_id": TELEGRAM_CHAT_ID, "text": text, "parse_mode": "Markdown" }
response = requests.post(url, json=payload, timeout=5) if not response.ok: raise RuntimeError( f"Telegram API error {response.status_code}: {response.text}" )
Gitリポジトリに保存され続けるのを防ぐため、秘密の環境変数を定義している点に注意してください。
そしてイベントコールバックから呼び出します：
from your_telegram_module import send_telegram_message
def my_callback(data): send_telegram_message("Asset created %s" % data"asset_id")
イベントリスナーをテストするには：
TELEGRAM_BOT_TOKEN=<TELEGRAM_BOT_TOKEN> TELEGRAM_CHAT_ID=<CHAT_ID> python server.py
4. カスタムKitsu APIエンドポイントでメッセージを受け取る
通知は便利ですが、双方向のコミュニケーションこそが、この統合を本当に価値あるものにします。
そのためには、Kitsuバックエンドを拡張して、/plugins/telegram/webhook のような新しいルートを登録するカスタムプラグインを用意する必要があります。詳細な手順は、公式の「Kitsuプラグインを開発する」ガイドを参照してください。
マニフェストは次のようになります：
id = "telegram"
name = "Telegram Bot"
description = "Telegram Bot"
version = "0.1.0"
maintainer = "Frank Rousseau <frank@cg-wire.com>"
website = "kitsu.cloud"
license = "AGPL-3.0-only"
maintainer_name = "Frank Rousseau"
maintainer_email = "frank@cg-wire.com"
frontend_project_enabled = true
frontend_studio_enabled = true
icon = "telegram"
そしてカスタムルートは、受け取ったコマンドを解析して、それを明確なバックエンドのアクションへマッピングします：
from flask_restful import Resourceclass WebhookResource(Resource): def post(self): args = self.get_args( ("message", {}, True), ("chat", {}, True), )
message = args['message'] chat_id = args['chat'].get("id") text = message.get("text", "") if text == "/hello": send_telegram_message("it works") return jsonify({"status": "ok"})
シンプルさのために /hello という単一のコマンドを定義していますが、さらにたくさん作って、Kitsuのサービスを使って制作データを問い合わせることもできます。
決定論的（deterministic）なコマンドは、テストしやすく、ログもしやすく、セキュアです。さらに一歩進めて、LLMに自然言語のリクエストをコマンドへマッピングさせることもできます。
必要なのは、主要なエントリポイント init.py にルートを登録することだけです：
from . import resources
routes = (f"/telegram/webhook", resources.WebhookResource)
Kitsuサーバーインスタンスにプラグインをパッケージングしてインストールしたら、次はTelegramボットに「どうやってそこへ到達するか」を教えます。
ローカル開発環境を使う場合、トンネルでサーバーを公開できます。たとえばngrokで、サーバーがポート5000で動いているなら：
ngrok http 5000
そして、TelegramボットのWebhookをそのURLへ向けるように設定します：
curl -X POST "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook"
-H "Content-Type: application/json"
-d '{"url": "https://<random>.ngrok-free.app/plugin/telegram/webhook"}'
これで、Telegramチャットでボットに /hello を送って結果を確認します：
結論
Kitsuと連携するカスタム・メッセージング統合は、常に似たパターンに従います。メッセージングプラットフォーム上でボットを作成し、Kitsuのイベントを購読し、構造化された通知を送信し、受信したメッセージを処理するためのバックエンドのルートを公開します。
しかし、それだけではありません。Kitsuプラグインにビューを追加することも検討してください。
たとえば、ボットのアクティビティや最近のやり取りをダッシュボード上に直接表示します。Kitsu内で作業するスーパーバイザーは、どのアラートが送られ、どのコマンドがトリガーされたかを確認できるようになります。可能性は無限です！
